package com.pureut.storage.util;

import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.pureut.storage.domain.PublicGoods;
import com.pureut.storage.domain.order.domain.ConverterManagement;
import com.pureut.storage.domain.order.domain.CustomerManagement;
import com.pureut.storage.domain.order.domain.OrderSupplierManagement;
import com.pureut.storage.dto.ConverterMqDto;
import com.pureut.storage.dto.CustomerManagementMqDto;
import com.pureut.storage.dto.OrderSupplierManagementMqDto;
import com.pureut.storage.mapper.ConverterManagementMapper;
import com.pureut.storage.mapper.CustomerManagementMapper;
import com.pureut.storage.mapper.OrderSupplierManagementMapper;
import com.pureut.storage.mapper.PublicGoodsMapper;
import com.pureut.storage.service.ConverterManagementService;
import com.pureut.storage.service.CustomerManagementService;
import com.pureut.storage.service.OrderSupplierManagementService;
import com.pureut.storage.service.PublicGoodsService;
import com.pureut.storage.vo.MaterialMqVo;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

/**
 * @Description mq订阅消息
 * @Date 22-12-22
 * @Author hl
 **/
@Service
public class RocketMqSubscribe {

    @Value("${rocketmq.name-server}")
    private String mqServer;

    @Value("${producer.group}")
    private String mqGroup;

    @Value("${producer.send-message-timeout}")
    private int mqTimeOut;

    /**
     * 公共物料表mapper
     */
    @Resource
    private PublicGoodsMapper publicGoodsMapper;

    /**
     * 公共物料表service
     */
    @Resource
    private PublicGoodsService publicGoodsService;

    /**
     * 加工商service
     */
    @Resource
    private ConverterManagementService converterManagementService;

    /**
     * 加工商mapper
     */
    @Resource
    private ConverterManagementMapper converterManagementMapper;

    /**
     * 供应商service
     */
    @Resource
    private OrderSupplierManagementService orderSupplierManagementService;

    /**
     * 供应商mapper
     */
    @Resource
    private OrderSupplierManagementMapper orderSupplierManagementMapper;

    /**
     * 客户service
     */
    @Resource
    private CustomerManagementService customerManagementService;

    /**
     * 客户mapper
     */
    @Resource
    private CustomerManagementMapper customerManagementMapper;

    /**
     * 订阅修改物料
     *
     * @throws MQClientException
     */
    @PostConstruct
    public void rocketMqSubscribe() throws MQClientException {
        //0.创造消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("material");
        //1.给消费者设置名称服务器
        consumer.setNamesrvAddr(mqServer);
        //2.订阅的主题和子表达式是什么
        consumer.subscribe(mqGroup, "*");
        //3.设置从哪里开始消费消息
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 设置消费模型，集群还是广播，默认为集群  CLUSTERING-集群   BROADCASTING-广播
        consumer.setMessageModel(MessageModel.BROADCASTING);
        //4.注册消息监听器,参数是并发消息监听器，Concurrently并发的意思
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                //逐条消费消息
                for (MessageExt msg : msgs) {
                    //输出消息的消费时间
                    System.out.println(new SimpleDateFormat("HH:mm:ss").format(new Date()));
                    System.out.println("," + new String(msg.getBody()));
                    String syn = new String(msg.getBody());
                    MaterialMqVo mq = JSON.parseObject(syn, MaterialMqVo.class);
                    if (mq.getMark() != 3) {
                        //通过物料编码获取数据，如果获取不到就是新增，如果获取到就是修改
                        PublicGoods publicGoods = publicGoodsMapper.getPublicGoodsByCode(mq.getMaterialCode());
                        if (publicGoods == null) {
                            //新增
                            publicGoodsService.insertPublicGoods(mq);
                        } else {
                            //修改
                            publicGoodsService.updatePublicGoods(mq);
                        }
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.开启消费者
        consumer.start();
        System.out.println("消费者已开启消费---");
    }

    /**
     * 订阅修改加工商
     *
     * @throws MQClientException
     */
    @PostConstruct
    public void rocketMqConverterManagement() throws MQClientException {
        //0.创造消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("converter_management");
        //1.给消费者设置名称服务器
        consumer.setNamesrvAddr(mqServer);
        //2.订阅的主题和子表达式是什么
        consumer.subscribe(mqGroup, "*");
        //3.设置从哪里开始消费消息
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 设置消费模型，集群还是广播，默认为集群  CLUSTERING-集群   BROADCASTING-广播
        consumer.setMessageModel(MessageModel.BROADCASTING);
        //4.注册消息监听器,参数是并发消息监听器，Concurrently并发的意思
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                //逐条消费消息
                for (MessageExt msg : msgs) {
                    //输出消息的消费时间
                    System.out.println(new SimpleDateFormat("HH:mm:ss").format(new Date()));
                    System.out.println("," + new String(msg.getBody()));
                    String syn = new String(msg.getBody());
                    ConverterMqDto mq = JSON.parseObject(syn, ConverterMqDto.class);
                    //通过加工商编码获取数据，如果获取不到就是新增，如果获取到就是修改
                    ConverterManagement converterManagement = converterManagementMapper.selectOne(new QueryWrapper<ConverterManagement>().lambda().eq(ConverterManagement::getConverterNum, mq.getConverterNum()));
                    if (converterManagement == null) {
                        //新增
                        converterManagementService.insertConverterManagement(mq);
                    } else {
                        //修改
                        converterManagementService.updateConverterManagement(mq);
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.开启消费者
        consumer.start();
        System.out.println("消费者已开启消费---");
    }

    /**
     * 订阅修改供应商
     *
     * @throws MQClientException
     */
    @PostConstruct
    public void rocketMqOrderSupplierManagement() throws MQClientException {
        //0.创造消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order_supplier_management");
        //1.给消费者设置名称服务器
        consumer.setNamesrvAddr(mqServer);
        //2.订阅的主题和子表达式是什么
        consumer.subscribe(mqGroup, "*");
        //3.设置从哪里开始消费消息
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 设置消费模型，集群还是广播，默认为集群  CLUSTERING-集群   BROADCASTING-广播
        consumer.setMessageModel(MessageModel.BROADCASTING);
        //4.注册消息监听器,参数是并发消息监听器，Concurrently并发的意思
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                //逐条消费消息
                for (MessageExt msg : msgs) {
                    //输出消息的消费时间
                    System.out.println(new SimpleDateFormat("HH:mm:ss").format(new Date()));
                    System.out.println("," + new String(msg.getBody()));
                    String syn = new String(msg.getBody());
                    OrderSupplierManagementMqDto mq = JSON.parseObject(syn, OrderSupplierManagementMqDto.class);
                    //通过供应商编码获取数据，如果获取不到就是新增，如果获取到就是修改
                    OrderSupplierManagement orderSupplierManagement = orderSupplierManagementMapper.selectOne(new QueryWrapper<OrderSupplierManagement>().lambda().eq(OrderSupplierManagement::getSupplierNum, mq.getSupplierNum()));
                    if (orderSupplierManagement == null) {
                        //新增
                        orderSupplierManagementService.insertOrderSupplierManagement(mq);
                    } else {
                        //修改
                        orderSupplierManagementService.updateOrderSupplierManagement(mq);
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.开启消费者
        consumer.start();
        System.out.println("消费者已开启消费---");
    }

    /**
     * 订阅修改客户
     *
     * @throws MQClientException
     */
    @PostConstruct
    public void rocketMqOrderCustomerManagement() throws MQClientException {
        //0.创造消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("customer_management");
        //1.给消费者设置名称服务器
        consumer.setNamesrvAddr(mqServer);
        //2.订阅的主题和子表达式是什么
        consumer.subscribe(mqGroup, "*");
        //3.设置从哪里开始消费消息
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 设置消费模型，集群还是广播，默认为集群  CLUSTERING-集群   BROADCASTING-广播
        consumer.setMessageModel(MessageModel.BROADCASTING);
        //4.注册消息监听器,参数是并发消息监听器，Concurrently并发的意思
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                //逐条消费消息
                for (MessageExt msg : msgs) {
                    //输出消息的消费时间
                    System.out.println(new SimpleDateFormat("HH:mm:ss").format(new Date()));
                    System.out.println("," + new String(msg.getBody()));
                    String syn = new String(msg.getBody());
                    CustomerManagementMqDto mq = JSON.parseObject(syn, CustomerManagementMqDto.class);
                    //通过客户编码获取数据，如果获取不到就是新增，如果获取到就是修改
                    CustomerManagement customerManagement = customerManagementMapper.selectOne(new QueryWrapper<CustomerManagement>().lambda().eq(CustomerManagement::getCustomerNum, mq.getCustomerNum()));
                    if (customerManagement == null) {
                        //新增
                        customerManagementService.insertCustomerManagement(mq);
                    } else {
                        //修改
                        customerManagementService.updateCustomerManagement(mq);
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.开启消费者
        consumer.start();
        System.out.println("消费者已开启消费---");
    }

    /**
     * mq删除物料信息
     *
     * @throws MQClientException
     */
    @PostConstruct
    public void rocketMqSubscribeDelete() throws MQClientException {
        //0.创造消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("materialDelete");
        //1.给消费者设置名称服务器
        consumer.setNamesrvAddr(mqServer);
        //2.订阅的主题和子表达式是什么
        consumer.subscribe(mqGroup, "*");
        //3.设置从哪里开始消费消息
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 设置消费模型，集群还是广播，默认为集群  CLUSTERING-集群   BROADCASTING-广播
        consumer.setMessageModel(MessageModel.BROADCASTING);
        //4.注册消息监听器,参数是并发消息监听器，Concurrently并发的意思
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                //逐条消费消息
                for (MessageExt msg : msgs) {
                    //输出消息的消费时间
                    System.out.println(new SimpleDateFormat("HH:mm:ss").format(new Date()));
                    System.out.println("," + new String(msg.getBody()));
                    String syn = new String(msg.getBody());
                    MaterialMqVo mq = JSON.parseObject(syn, MaterialMqVo.class);
                    if (mq.getMark() == 3) {
                        //删除公共物料信息
                        publicGoodsService.deletePublicGoods(mq);
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.开启消费者
        consumer.start();
        System.out.println("消费者已开启消费---");
    }

    /**
     * mq删除客户信息
     *
     * @throws MQClientException
     */
    @PostConstruct
    public void rocketMqCustomerDelete() throws MQClientException {
        //0.创造消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("customer_management_delete");
        //1.给消费者设置名称服务器
        consumer.setNamesrvAddr(mqServer);
        //2.订阅的主题和子表达式是什么
        consumer.subscribe(mqGroup, "*");
        //3.设置从哪里开始消费消息
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 设置消费模型，集群还是广播，默认为集群  CLUSTERING-集群   BROADCASTING-广播
        consumer.setMessageModel(MessageModel.BROADCASTING);
        //4.注册消息监听器,参数是并发消息监听器，Concurrently并发的意思
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                //逐条消费消息
                for (MessageExt msg : msgs) {
                    //输出消息的消费时间
                    System.out.println(new SimpleDateFormat("HH:mm:ss").format(new Date()));
                    System.out.println("," + new String(msg.getBody()));
                    String syn = new String(msg.getBody());
                    CustomerManagementMqDto mq = JSON.parseObject(syn, CustomerManagementMqDto.class);
                    if (mq.getCustomerDeleteMark() == 13) {
                        //删除客户
                        customerManagementService.deleteCustomerManagement(mq);
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.开启消费者
        consumer.start();
        System.out.println("消费者已开启消费---");
    }

    /**
     * mq删除加工商信息
     *
     * @throws MQClientException
     */
    @PostConstruct
    public void rocketMqConverterDelete() throws MQClientException {
        //0.创造消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("converter_management_delete");
        //1.给消费者设置名称服务器
        consumer.setNamesrvAddr(mqServer);
        //2.订阅的主题和子表达式是什么
        consumer.subscribe(mqGroup, "*");
        //3.设置从哪里开始消费消息
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 设置消费模型，集群还是广播，默认为集群  CLUSTERING-集群   BROADCASTING-广播
        consumer.setMessageModel(MessageModel.BROADCASTING);
        //4.注册消息监听器,参数是并发消息监听器，Concurrently并发的意思
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                //逐条消费消息
                for (MessageExt msg : msgs) {
                    //输出消息的消费时间
                    System.out.println(new SimpleDateFormat("HH:mm:ss").format(new Date()));
                    System.out.println("," + new String(msg.getBody()));
                    String syn = new String(msg.getBody());
                    ConverterMqDto mq = JSON.parseObject(syn, ConverterMqDto.class);
                    if (mq.getConverterDeleteMark() == 11) {
                        //删除加工商
                        converterManagementService.deleteConverterManagement(mq);
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.开启消费者
        consumer.start();
        System.out.println("消费者已开启消费---");
    }

    /**
     * mq删除供应商信息
     *
     * @throws MQClientException
     */
    @PostConstruct
    public void rocketMqOrderSupplierDelete() throws MQClientException {
        //0.创造消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order_supplier_management_delete");
        //1.给消费者设置名称服务器
        consumer.setNamesrvAddr(mqServer);
        //2.订阅的主题和子表达式是什么
        consumer.subscribe(mqGroup, "*");
        //3.设置从哪里开始消费消息
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 设置消费模型，集群还是广播，默认为集群  CLUSTERING-集群   BROADCASTING-广播
        consumer.setMessageModel(MessageModel.BROADCASTING);
        //4.注册消息监听器,参数是并发消息监听器，Concurrently并发的意思
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                //逐条消费消息
                for (MessageExt msg : msgs) {
                    //输出消息的消费时间
                    System.out.println(new SimpleDateFormat("HH:mm:ss").format(new Date()));
                    System.out.println("," + new String(msg.getBody()));
                    String syn = new String(msg.getBody());
                    OrderSupplierManagementMqDto mq = JSON.parseObject(syn, OrderSupplierManagementMqDto.class);
                    if (mq.getSupplierDeleteMark() == 12) {
                        //删除加工商
                        orderSupplierManagementService.deleteOrderSupplierManagement(mq);
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.开启消费者
        consumer.start();
        System.out.println("消费者已开启消费---");
    }
}
